package cn.wanda.utils.db

import cn.wanda.utils.config.groupJobTrait
import cn.wanda.utils.modelGroup
import org.apache.spark.sql.SparkSession

import scala.collection.mutable

class CreateBigTable(sparkSessionContext: SparkSession, appParaMap: mutable.HashMap[String, String]) extends groupJobTrait {
    val url1 = appParaMap("oracle.url1")
    val username1 = appParaMap("oracle.username1")
    val password1 = appParaMap("oracle.password1")
    val databases1 = appParaMap("oracle.databases1")

    val url2 = appParaMap("oracle.url2")
    val username2 = appParaMap("oracle.username2")
    val password2 = appParaMap("oracle.password2")
    val databases2 = appParaMap("oracle.databases2")
    val oracle = new OracleDBManager

    /**
      * 替换表内库名和表名
      *
      * @param sql
      * @return
      */
    def replaceTableAndDatabase(sql: String): String = {
        sql
          .replaceAll("wise", databases1)
          .replaceAll("frms_portal_credit", databases2)
          .replaceAll("lm_installment_tran ", appParaMap("lm_installment_tran"))
          .replaceAll("ts_installment_main", appParaMap("ts_installment_main"))
          .replaceAll("lm_installment_tran_d", appParaMap("lm_installment_tran_d"))
          .replaceAll("ts_installment_detail", appParaMap("ts_installment_detail"))
          .replaceAll("fcs_acct_register", appParaMap("fcs_acct_register"))
          .replaceAll("fcs_txn", appParaMap("fcs_txn"))
          .replaceAll("lm_coustomer", appParaMap("lm_coustomer"))
          .replaceAll("rams_cdt_checklist ", appParaMap("rams_cdt_checklist"))
          .replaceAll("rams_cdt_checklist_risks", appParaMap("rams_cdt_checklist_risks"))
          .replaceAll("sysdate", "sysdate - 195")
        // TODO sysdate测试使用
    }

    /**
      * 从远程oracle读取合并关键表，并生成临时表
      */
    def get_whmd_hisdue1_11(): Unit = {
//        val sql = replaceTableAndDatabase(whmd_hisdue1_11)
        val sql = replaceTableAndDatabase(whmd_hisdue1_11_2)
        oracle.registTmpFromOracle(sparkSessionContext, sql, url1, username1, password1, "whmd_hisdue1_11")
    }

    def get_namelist(): Unit = {
//        val sql = replaceTableAndDatabase(namelist)
        val sql = "(select * from wise.namelist_2) t"
        oracle.registTmpFromOracle(sparkSessionContext, sql, url1, username1, password1, "namelist")
    }

    /**
      * 生成中间临时表
      */
    def create_whmd_hisdue(): Unit = {
        sparkSessionContext.sql(whmd_hisdue).toDF().createOrReplaceTempView("whmd_hisdue")
    }

    def create_checklist_tmp(): Unit = {
        val mem_rows = sparkSessionContext.sql("select member_code from namelist").select("member_code").distinct().collectAsList()
        val result = mem_rows.toArray.mkString("'", "', '", "'").replaceAll("\\[", "").replaceAll("\\]", "")

        oracle.registTmpFromOracle(sparkSessionContext, replaceTableAndDatabase(checklistTmpSql(result)), url2, username2, password2, "checklist_tmp")
    }

    def create_checklist_tmp2(): Unit = {
        sparkSessionContext.sql(checklist_tmp2).toDF().createOrReplaceTempView("checklist_tmp2")
    }

    def create_checklist_tmp3(): Unit = {
        sparkSessionContext.sql(checklist_tmp3).toDF().createOrReplaceTempView("checklist_tmp3")
    }

    def create_checklist_tmp4(): Unit = {
        sparkSessionContext.sql(checklist_tmp4).toDF().createOrReplaceTempView("checklist_tmp4")
    }

    def create_whmd_tran_tmp(): Unit = {
        sparkSessionContext.sql(whmd_tran_tmp).toDF().createOrReplaceTempView("whmd_tran_tmp")
    }

    def create_whmd_tran_tmp2(): Unit = {
        sparkSessionContext.sql(whmd_tran_tmp2).toDF().createOrReplaceTempView("whmd_tran_tmp2")
    }

    def create_A_shidong11(): Unit = {
        sparkSessionContext.sql(A_shidong11).toDF().createOrReplaceTempView("A_shidong11")
    }

    def create_whmd_shouyu(): Unit = {
        sparkSessionContext.sql(whmd_shouyu).toDF().createOrReplaceTempView("whmd_shouyu")
    }

    def create_whmd_tran_var(): Unit = {
        sparkSessionContext.sql(whmd_tran_var).toDF().createOrReplaceTempView("whmd_tran_var")
    }

    def create_tran_var_finl(): Unit = {
        sparkSessionContext.sql(tran_var_finl).toDF().createOrReplaceTempView("tran_var_finl")
    }

    def create_whmd_tran_tmp3(): Unit = {
        sparkSessionContext.sql(whmd_tran_tmp3).toDF().createOrReplaceTempView("whmd_tran_tmp3")
    }

    def create_A_shidong(): Unit = {
        sparkSessionContext.sql(A_shidong).toDF().createOrReplaceTempView("A_shidong")
    }

    def create_A_day1(): Unit = {
        sparkSessionContext.sql(A_day1).toDF().createOrReplaceTempView("A_day1")
    }

    def create_A_day(): Unit = {
        sparkSessionContext.sql(A_day).toDF().createOrReplaceTempView("A_day")
    }

    def create_lst_txn_dtl(): Unit = {
        sparkSessionContext.sql(lst_txn_dtl).toDF().createOrReplaceTempView("lst_txn_dtl")
    }

    def create_lst_txn_var(): Unit = {
        sparkSessionContext.sql(lst_txn_var).toDF().createOrReplaceTempView("lst_txn_var")
    }

    def create_ccard_bigtable11(): Unit = {
        sparkSessionContext.sql(ccard_bigtable11).toDF().createOrReplaceTempView("ccard_bigtable11")
    }

    /**
      * 生成最终大表
      */
    def createTable(): Unit = {

        get_namelist()
        get_whmd_hisdue1_11()
        (new modelGroup.createOdIntvlMax).create(sparkSessionContext)
        (new modelGroup.createAvgShiDong).create(sparkSessionContext)
        create_whmd_hisdue()
        create_checklist_tmp()
        create_checklist_tmp2()
        create_checklist_tmp3()
        create_checklist_tmp4()
        create_whmd_tran_tmp()
        create_whmd_tran_tmp2()
        create_A_shidong11()
        create_whmd_shouyu()
        create_whmd_tran_var()
        create_tran_var_finl()
        create_whmd_tran_tmp3()
        create_A_shidong()
        create_A_day1()
        create_A_day()
        create_lst_txn_dtl()
        create_lst_txn_var()
        create_ccard_bigtable11()

        /**
          * 清空内存临时表，只保留大表
          */
        sparkSessionContext.sql("drop table whmd_hisdue1_11")
        sparkSessionContext.sql("drop table od_intvl_max")
        sparkSessionContext.sql("drop table shidong_avg_3mon")
        sparkSessionContext.sql("drop table whmd_hisdue")
        sparkSessionContext.sql("drop table checklist_tmp")
        sparkSessionContext.sql("drop table checklist_tmp2")
        sparkSessionContext.sql("drop table checklist_tmp3")
        sparkSessionContext.sql("drop table checklist_tmp4")
        sparkSessionContext.sql("drop table whmd_tran_tmp")
        sparkSessionContext.sql("drop table whmd_tran_tmp2")
        sparkSessionContext.sql("drop table A_shidong11")
        sparkSessionContext.sql("drop table whmd_shouyu")
        sparkSessionContext.sql("drop table whmd_tran_var")
        sparkSessionContext.sql("drop table tran_var_finl")
        sparkSessionContext.sql("drop table whmd_tran_tmp3")
        sparkSessionContext.sql("drop table A_shidong")
        sparkSessionContext.sql("drop table A_day1")
        sparkSessionContext.sql("drop table A_day")
        sparkSessionContext.sql("drop table lst_txn_dtl")
        sparkSessionContext.sql("drop table lst_txn_var")

    }

}
